/*
 * meituan.com Inc.
 * Copyright (c) 2010-2018 All Rights Reserved.
 */
package java.zk.example;

import com.alibaba.fastjson.JSON;
import org.apache.zookeeper.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.Random;

/**
 * <p>
 * ZK客户端
 * </p>
 * @author LvJing
 * @version $Id:Client.java v1.0 2018/9/2 上午11:43 LvJing Exp $
 */
public class Client implements Watcher {

    private static final Logger log = LoggerFactory.getLogger(Client.class);

    private ZooKeeper zk;

    private String hostPort;

    private String serverId = Integer.toHexString(new Random().nextInt());

    public Client(String hostPort) {
        this.hostPort = hostPort;
    }

    public void startZK() throws IOException {
        this.zk = new ZooKeeper(hostPort, 150000, this);
    }

    /**
     * 创建序列号任务
     */
    public String queueCommand(String command) throws KeeperException, InterruptedException {
        while (true) {
            try {
                return zk.create("/tasks/task-", command.getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE,
                        CreateMode.PERSISTENT_SEQUENTIAL);
            } catch (KeeperException.NodeExistsException e) {
                // 正常情况下不会出现创建的序列号重复的情况，如果出现，那说明该序列化机制出现bug
                throw new RuntimeException("execute " + command + ", appears to be running");
            } catch (KeeperException.ConnectionLossException e) {
                // 连接丢失异常，忽略，再次重试
                log.warn("connection loss, try again");
            }
        }
    }

    /**
     * 客户端提交任务到ZK上，在/tasks/task-id创建任务，id为有序znode。
     * 提交成功后监听/status/task-id变化情况
     * @param taskData 任务需传递的数据
     */
    public void submitTask(TaskData taskData) {
        zk.create("/tasks/task-", JSON.toJSONBytes(taskData), ZooDefs.Ids.OPEN_ACL_UNSAFE,
                CreateMode.PERSISTENT_SEQUENTIAL, (rc, path, ctx, name) -> {
                    switch (KeeperException.Code.get(rc)) {
                        case CONNECTIONLOSS:
                            log.warn("{} 连接丢失，检查是否提交任务成功");
                            // TODO: 2018/9/4 遍历/tasks下所有任务，看是否有那个的task的serverId与当前相等的。这样只能变成，提交一个任务
                            break;
                        case OK:
                            log.info("任务提交成功, taskId:{}", name);
                            watchTaskStatus(name);
                            break;
                        default:
                            log.error("{} 提交任务时失败", serverId, KeeperException.create(KeeperException.Code.get(rc)));
                    }
                }, taskData);
    }

    /**
     * 监视任务执行状态
     */
    private void watchTaskStatus(String taskId) {
        zk.exists("/status/" + taskId, event -> {
            if (event.getType().equals(Event.EventType.NodeCreated)) {
                zk.getData("/status/" + taskId, false,
                        (rc1, path1, ctx1, data, stat1) -> log.info("任务{}执行完成", path1), null);
            }
        }, (rc, path, ctx, stat) -> {
            switch (KeeperException.Code.get(rc)) {
                case CONNECTIONLOSS:
                    log.warn("检查/status/{}节点是否存在时连接丢失，再次重试", taskId);
                    watchTaskStatus(taskId);
                    break;
                case NONODE:
                    break;
                case OK:
                    zk.getData("/status/" + taskId, false,
                            (rc1, path1, ctx1, data, stat1) -> log.info("任务{}执行完成", path1), null);
                    break;
                default:
                    log.error("检查任务状态节点时发生错误, taskId:{}", taskId, KeeperException.create(KeeperException.Code.get(rc)));
            }
        }, taskId);
    }

    @Override
    public void process(WatchedEvent event) {
        log.info("WatchedEvent:{}, hostPort:{}", event, hostPort);
    }

    public static void main(String[] args) throws Exception {
        Client client = new Client("127.0.0.1:2181");
        client.startZK();
        client.submitTask(new TaskData(client.serverId, "我是任务名称"));
        System.out.println(client.queueCommand("test"));
    }
}
